"""
版本说明：基于异步事件驱动的开仓逻辑实现
使用异步事件驱动模式管理订单生命周期，提高系统的并发处理能力和响应性

和上一版本区别：
    上一版本文件名为：Server_MySql_Only_01.py
    本版本使用异步事件驱动模式重新实现开仓逻辑，替代同步等待机制
"""

import time, os, datetime
from threading import Thread, Lock
from dbutils.pooled_db import PooledDB
import pymysql
import numpy as np
import pandas as pd
from tqsdk import TqApi, TqAuth, TqKq
from rich.console import Console
from rich.theme import Theme
from MyTT import REF, EMA, SUM
import logging
import json

# 异步订单管理器类
class OrderManager:
    def __init__(self):
        self.orders = {}  # order_id -> order_info
        self.callbacks = {}  # order_id -> callback functions
        self.lock = Lock()
        
    def add_order(self, order_id, symbol, direction, offset, volume, limit_price, description):
        """添加新订单到异步管理"""
        with self.lock:
            order_info = {
                'order_id': order_id,
                'symbol': symbol,
                'direction': direction,
                'offset': offset,
                'volume': volume,
                'limit_price': limit_price,
                'description': description,
                'status': 'PENDING',
                'create_time': time.time(),
                'last_update_time': time.time()
            }
            self.orders[order_id] = order_info
            
            console.print(
                f"{time.strftime('%X')} {symbol}{description}: 手数: {volume}, 订单ID: {order_id}",
                style="trade"
            )
            logging.info(f"{symbol}{description}: 手数: {volume}, 价格: {limit_price}, 订单ID: {order_id}")
            
    def register_callback(self, order_id, callback):
        """注册订单状态变化回调函数"""
        with self.lock:
            self.callbacks[order_id] = callback
            
    def update_order_status(self, api):
        """异步更新所有订单状态"""
        with self.lock:
            orders_to_remove = []
            
            for order_id, order_info in self.orders.items():
                try:
                    order = api.get_order(order_id)
                    if order:
                        old_status = order_info['status']
                        
                        if order.status == 'FINISHED':
                            order_info['status'] = 'FILLED'
                            order_info['filled_price'] = order.price if hasattr(order, 'price') else order_info['limit_price']
                            self._on_order_filled(order_info)
                            
                        elif order.status == 'ALIVE':
                            order_info['status'] = 'SUBMITTED'
                            # 检查是否超时（5秒）
                            if time.time() - order_info['create_time'] > 5:
                                order_info['status'] = 'TIMEOUT'
                                self._on_order_timeout(order_info, api)
                                
                        elif order.status == 'CANCELLED':
                            order_info['status'] = 'CANCELLED'
                            self._on_order_cancelled(order_info)
                            
                        # 触发回调函数
                        if old_status != order_info['status'] and order_id in self.callbacks:
                            try:
                                self.callbacks[order_id](order_info)
                            except Exception as e:
                                logging.error(f"订单回调函数执行失败: {e}")
                                
                        # 如果订单已经完成，标记为待移除
                        if order_info['status'] in ['FILLED', 'CANCELLED', 'TIMEOUT']:
                            orders_to_remove.append(order_id)
                            
                except Exception as e:
                    logging.error(f"更新订单状态失败: {e}")
            
            # 移除已完成订单
            for order_id in orders_to_remove:
                if order_id in self.orders:
                    del self.orders[order_id]
                if order_id in self.callbacks:
                    del self.callbacks[order_id]
                    
    def _on_order_filled(self, order_info):
        """订单成交处理"""
        console.print(
            f"{time.strftime('%X')} {order_info['symbol']}{order_info['description']}订单已成交",
            style="success"
        )
        logging.info(f"{order_info['symbol']}{order_info['description']}订单已成交: {order_info['order_id']}")
        
        # 记录交易到数据库
        insert_trade_record(
            order_info['symbol'], 
            order_info['direction'], 
            order_info['offset'], 
            order_info['volume'], 
            order_info.get('filled_price', order_info['limit_price']), 
            order_info['description']
        )
        
    def _on_order_cancelled(self, order_info):
        """订单撤销处理"""
        console.print(
            f"{time.strftime('%X')} {order_info['symbol']}{order_info['description']}订单已撤销",
            style="warning"
        )
        logging.info(f"{order_info['symbol']}{order_info['description']}订单已撤销: {order_info['order_id']}")
        
    def _on_order_timeout(self, order_info, api):
        """订单超时处理"""
        console.print(
            f"{time.strftime('%X')} {order_info['symbol']}{order_info['description']}订单超时未成交",
            style="warning"
        )
        logging.info(f"{order_info['symbol']}{order_info['description']}订单超时未成交: {order_info['order_id']}")
        
        # 尝试撤销订单
        try:
            api.cancel_order(order_info['order_id'])
            console.print(
                f"{time.strftime('%X')} {order_info['symbol']}{order_info['description']}订单超时，已撤销",
                style="warning"
            )
            logging.info(f"{order_info['symbol']}{order_info['description']}订单超时，已撤销: {order_info['order_id']}")
        except Exception as e:
            logging.error(f"撤销超时订单失败: {e}")
            
    def get_active_orders_count(self):
        """获取活跃订单数量"""
        with self.lock:
            return len([order for order in self.orders.values() 
                       if order['status'] in ['PENDING', 'SUBMITTED']])

# 加载数据库配置
def load_db_config():
    """
    从db_config.json文件加载数据库配置
    :return: 数据库配置字典
    """
    config_path = os.path.join(os.path.dirname(__file__), 'db_config.json')
    # 定义统一的默认配置
    default_config = {
        "db_type": "mysql",
        "sqlite_path": "trade_monitor.db",
        "mysql_host": "172.28.196.123",
        "mysql_port": 3306,
        "mysql_user": "remote_admin",
        "mysql_password": "alibaba999",
        "mysql_db_name": "trade_monitor"
    }
    
    try:
        if os.path.exists(config_path):
            with open(config_path, 'r', encoding='utf-8') as f:
                config = json.load(f)
                return config
        else:
            # 如果配置文件不存在，返回默认配置
            console.print("配置文件db_config.json不存在，使用默认配置", style="warning")
            return default_config
    except Exception as e:
        console.print(f"读取数据库配置文件时出错: {e}", style="error")
        logging.error(f"读取数据库配置文件时出错: {e}", exc_info=True)
        # 读取错误时也返回统一的默认配置
        return default_config

# 加载数据库配置
DB_CONFIG = load_db_config()

# 定义控制台输出，更明亮显眼颜色主题
custom_theme = Theme({
    "info": "bright_cyan",
    "warning": "bright_yellow",
    "error": "bright_red",
    "success": "bright_green",
    "trade": "bold bright_magenta"
})
console = Console(theme=custom_theme)

# MySQL数据库连接池
db = None
try:
    mysql_config = {
        'host': DB_CONFIG.get('mysql_host', 'localhost'),
        'port': DB_CONFIG.get('mysql_port', 3306),
        'user': DB_CONFIG.get('mysql_user', 'remote_admin'),
        'password': DB_CONFIG.get('mysql_password', 'Admin@szlwy68'),
        'database': DB_CONFIG.get('mysql_db_name', 'trade_monitor')
    }
    
    console.print(f"尝试连接到MySQL数据库: {mysql_config['host']}:{mysql_config['port']}", style="info")
    
    db = PooledDB(
        creator=pymysql,
        host=mysql_config['host'],
        port=mysql_config['port'],
        user=mysql_config['user'],
        password=mysql_config['password'],
        database=mysql_config['database'],
        maxconnections=5,
        mincached=2,
        maxcached=3,
        maxshared=3,
        blocking=True,
        autocommit=True,        # 添加自动提交选项
        charset='utf8mb4',      # 指定字符集
        connect_timeout=5,      # 连接超时时间，单位秒
    )
    
    # 测试连接
    conn = db.connection()
    cursor = conn.cursor()
    cursor.execute("SELECT 1")
    result = cursor.fetchone()
    cursor.close()
    conn.close()
    
    if result and result[0] == 1:
        console.print(f"成功连接到MySQL数据库: {mysql_config['host']}:{mysql_config['port']}", style="success")
    else:
        raise Exception("数据库连接测试失败")
        
except pymysql.MySQLError as e:
    # 捕获MySQL连接错误，显示红色友好提示
    console.print(f"无法连接到MySQL数据库: {str(e)}", style="error")
    console.print("请检查以下事项：", style="warning")
    console.print("1. MySQL服务器是否启动", style="warning")
    console.print("2. 网络连接是否正常", style="warning")
    console.print("3. 防火墙设置是否正确", style="warning")
    console.print("4. 数据库连接参数是否正确", style="warning")
    console.print("5. 用户权限是否足够", style="warning")
    logging.error(f"数据库连接失败: {str(e)}", exc_info=True)
    
    # 询问用户是否继续使用模拟模式
    console.print("\n由于数据库连接失败，程序无法继续运行。", style="error")
    console.print("请检查数据库配置并重新启动程序。", style="warning")
    exit(1)
    
except Exception as e:
    console.print(f"数据库连接过程中发生错误: {str(e)}", style="bold red")
    logging.error(f"数据库连接失败: {str(e)}", exc_info=True)
    console.print("程序无法继续运行，请检查配置后重试。", style="error")
    exit(1)

# 配置日志记录
logging.basicConfig(
    filename='trading_log.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

######################################## 定义全局变量########################################
symbols = []
api = None
threads = []
order_manager = OrderManager()  # 全局订单管理器

# 定义开仓资金比例（占可用资金的百分比）
POSITION_FUND_RATIO = 0.2  # 20%，可以根据需要调整

######################################## 定义函数########################################
# 新增函数：获取 config 表所有配置项
def get_all_configs():
    """
    从 config 表中获取所有配置项，以字典形式返回
    :return: 包含所有配置项的字典
    """
    try:
        conn = db.connection()
        cursor = conn.cursor()
        cursor.execute("SELECT `key`, `value` FROM config")
        results = cursor.fetchall()
        cursor.close()
        conn.close()
        
        # 创建配置字典
        config_dict = dict(results)
        return config_dict
    except Exception as e:
        console.print(f"从数据库获取所有配置项时出错: {e}", style="error")
        logging.error(f"从数据库获取所有配置项时出错: {e}", exc_info=True)
        return {}

def get_symbols():
    try:
        conn = db.connection()
        cursor = conn.cursor()
        # 查询品种代码和 mtp 字段
        cursor.execute("SELECT exchange_code FROM products WHERE is_monitored=1")
        products = cursor.fetchall()
        cursor.close()
        conn.close()
        
        # 返回包含品种代码和 mtp 值的列表
        return [product[0] for product in products]
    except Exception as e:
        console.print(f"从数据库获取监控品种时出错: {e}", style="error")
        logging.error(f"从数据库获取监控品种时出错: {e}", exc_info=True)
        return []

# 新增函数：记录交易到数据库
def insert_trade_record(symbol, direction, offset, volume, price, reason):
    conn = None
    try:
        timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
        # 获取当前时间作为created_at字段值
        created_at = datetime.datetime.now()
        
        # MySQL连接方式
        conn = db.connection()
        cursor = conn.cursor()
        
        # 使用参数化查询防止SQL注入 (MySQL使用%s作为占位符)
        sql = """
        INSERT INTO trade_records 
        (product_name, direction, offset, volume, price, reason, timestamp, created_at) 
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
        """
        cursor.execute(sql, (symbol, direction, offset, volume, price, reason, timestamp, created_at))
        conn.commit()
        cursor.close()
        
        logging.info(f"成功记录交易: {symbol} {direction} {offset} {volume}手 价格{price}")
    except Exception as e:
        console.print(f"记录交易到数据库时出错: {e}", style="error")
        logging.error(f"记录交易到数据库时出错: {e}", exc_info=True)
        # 发生异常时回滚事务
        if conn:
            try:
                conn.rollback()
            except Exception as rollback_error:
                logging.error(f"回滚事务时出错: {rollback_error}")
    finally:
        # 确保连接关闭
        if conn:
            try:
                conn.close()
            except Exception as e:
                logging.error(f"关闭数据库连接时出错: {e}")

# 检查开仓条件函数
def check_open_condition(symbol, configs, position):
    """
    检查是否可以开仓
    :return: (是否可以开仓, 原因说明)
    """
    try:
        # 检查是否在交易时间段内
        current_time = datetime.datetime.now().time()
        
        # 定义交易时间段（可以根据需要调整）
        trading_start = datetime.time(9, 0)   # 上午9:00
        trading_end = datetime.time(15, 0)   # 下午3:00
        
        if not (trading_start <= current_time <= trading_end):
            return False, "非交易时间段"
            
        # 检查是否已经有持仓
        if position.pos_long > 0 or position.pos_short > 0:
            return False, "已有持仓"
            
        # 检查开仓方向参数
        dks = configs.get('dks', '双向')
        if dks == "不开仓仅止损":
            return False, "开仓方向设置为不开仓仅止损"
            
        return True, "可以开仓"
        
    except Exception as e:
        logging.error(f"检查开仓条件时出错: {e}")
        return False, f"检查开仓条件时出错: {e}"

# 基于异步事件驱动的开仓函数
def open_position_async(symbol, direction, is_shfe=False, api_instance=None):
    """
    使用异步事件驱动模式开仓
    :param symbol: 品种代码
    :param direction: 方向 ("BUY" 或 "SELL")
    :param is_shfe: 是否为上海交易所品种
    :param api_instance: TqApi实例，如果为None则使用全局api
    :return: 订单ID，如果开仓失败返回None
    """
    try:
        # 获取配置
        configs = get_all_configs()
        
        # 检查自动开仓是否开启
        auto_open = configs.get('auto_open', 'off').lower() == 'on'
        if not auto_open:
            console.print(f"{time.strftime('%X')} 自动开仓功能已关闭，跳过开仓检查", style="warning")
            return None
            
        # 使用传入的api实例或全局api
        current_api = api_instance if api_instance is not None else api
        if current_api is None:
            console.print(f"{time.strftime('%X')} API实例未初始化，无法开仓", style="error")
            return None
            
        # 获取品种数据
        quote = current_api.get_quote(symbol)
        position = current_api.get_position(symbol)
        
        # 计算开仓价格
        if direction == "BUY":
            limit_price = quote.ask_price1 + 2 * quote.price_tick
            description = "多仓"
        else:  # SELL
            limit_price = quote.bid_price1 - 2 * quote.price_tick
            description = "空仓"
            
        # 计算开仓手数
        math_open_lots = configs.get('math_open_lots', 'off').lower() == 'on'
        volume = 1
        
        if math_open_lots:
            # 从数据库的products表获取当前品种每手保证金
            try:
                conn = db.connection()
                cursor = conn.cursor()
                cursor.execute('SELECT margin FROM products WHERE exchange_code = %s', (symbol,))
                product_bond = cursor.fetchone()
                if product_bond is None:
                    product_bond = (1000,)  # 设置默认保证金值
            except Exception as e:
                logging.error(f"MySQL查询失败: {e}")
                product_bond = (1000,)  # 设置默认保证金值
            finally:
                if 'cursor' in locals():
                    cursor.close()
                if 'conn' in locals():
                    try:
                        conn.close()
                    except Exception as e:
                        logging.error(f"关闭数据库连接失败: {e}")
                        
            if product_bond and product_bond[0]:
                # 获取可用资金
                available_funds = current_api.get_account().available
                # 按可用资金20%除以当前品种保证金后取整再除以3计算目标持仓量
                target_volume = int(((available_funds * POSITION_FUND_RATIO) / product_bond[0]) / 3)
                # 获取已有持仓量
                if direction == "BUY":
                    current_position = position.pos_long
                else:
                    current_position = position.pos_short
                # 计算需要开仓的手数
                try:
                    volume = max(1, target_volume - current_position)
                except Exception as e:
                    logging.error(f"计算开仓手数时出错: {e}")
                    volume = 1
        
        print(f"当前品种开仓手数是：{volume}")
        
        # 执行开仓操作
        order_id = current_api.insert_order(
            symbol=symbol,
            direction=direction,
            offset="OPEN",
            volume=volume,
            limit_price=limit_price
        )
        
        # 添加到异步订单管理器
        order_manager.add_order(order_id, symbol, direction, "OPEN", volume, limit_price, description)
        
        return order_id
        
    except Exception as e:
        console.print(f"开仓时出错: {e}", style="error")
        logging.error(f"开仓时出错: {e}", exc_info=True)
        return None

# 止损函数（基于异步事件驱动）
def stop_loss_async(symbol):
    """
    止损函数 - 使用异步事件驱动模式管理订单
    """
    # 每个线程创建自己的api实例，避免资源冲突
    try:
        thread_api = TqApi(auth=TqAuth("cps168", "alibaba"))
    except Exception as e:
        console.print(f"初始化TqApi失败: {e}", style="error")
        return
        
    # 获取品种的K线数据
    kline = thread_api.get_kline_serial(symbol, 60, 100)
    
    while True:
        try:
            thread_api.wait_update()
            
            # 异步更新所有订单状态
            order_manager.update_order_status(thread_api)
            
            # 获取当前品种的报价和持仓
            quote = thread_api.get_quote(symbol)
            position = thread_api.get_position(symbol)
            
            # 获取配置
            configs = get_all_configs()
            
            # 检查是否可以开仓
            can_open, reason = check_open_condition(symbol, configs, position)
            
            if can_open:
                # 获取开仓方向参数
                dks = configs.get('dks', '双向')
                
                # 计算技术指标
                trading_line = EMA(kline.close, 5)  # 操盘线
                short_line = EMA(kline.close, 20)   # 空头线
                
                # 检查是否需要开多仓 - 金叉条件
                if position.pos_long == 0 and (dks == "只做多" or dks == "双向") and dks != "只做空" and dks != "不开仓仅止损":
                    if len(trading_line) >= 3 and len(short_line) >= 3:
                        if trading_line[-2] > short_line[-2] and trading_line[-3] < short_line[-3]:
                            is_shfe = symbol.startswith('SHFE')
                            open_position_async(symbol, "BUY", is_shfe, thread_api)
                        
                # 检查是否需要开空仓 - 死叉条件
                if position.pos_short == 0 and (dks == "只做空" or dks == "双向") and dks != "只做多" and dks != "不开仓仅止损":
                    if len(trading_line) >= 3 and len(short_line) >= 3:
                        if trading_line[-2] < short_line[-2] and trading_line[-3] > short_line[-3]:
                            is_shfe = symbol.startswith('SHFE')
                            open_position_async(symbol, "SELL", is_shfe, thread_api)
                        
            else:
                console.print(f"{time.strftime('%X')} {symbol} 不可以开仓: {reason}", style="warning")
                
            # 短暂休眠，避免过于频繁的检查
            time.sleep(0.1)
            
        except Exception as e:
            console.print(f"处理 {symbol} 时出错: {e}", style="error")
            time.sleep(1)

####################################################主函数##############################################
if __name__ == "__main__":
    threads = []
    try:
        # 测试数据库连接
        try:
            conn = db.connection()
            cursor = conn.cursor()
            cursor.execute("SELECT 1")
            result = cursor.fetchone()
            cursor.close()
            conn.close()
            if result and result[0] == 1:
                console.print("MySQL数据库连接测试成功!", style="success")
            else:
                console.print("MySQL数据库连接测试失败!", style="error")
        except Exception as e:
            console.print(f"数据库连接测试失败: {e}", style="error")
            
        # 获取待监控的品种
        symbol_list = get_symbols()
        if not symbol_list:
            console.print("没有找到需要监控的品种，请检查数据库配置", style="warning")
            exit(1)
            
        console.print(f"**********{time.strftime('%X')} 系统开始运行 **********", style="warning")
        console.print(f"监控品种列表: {symbol_list}", style="info")
        
        i=1
        for symbol in symbol_list:
            console.print(f"**********{time.strftime('%X')} {i} ，开始监控 {symbol} **********", style="bold yellow")
            t = Thread(target=stop_loss_async, args=(symbol,), daemon=True)
            t.start()
            threads.append(t)
            time.sleep(3)
            i=i+1
            
        # 主线程等待所有子线程完成
        while True:
            time.sleep(10)
            
    except KeyboardInterrupt:
        console.print("正在停止所有线程...", style="bold yellow")
        if api is not None:
            api.close()
    except Exception as e:
        console.print(f"主程序出错: {e}", style="bold red")
        logging.error(f"主程序出错: {e}", exc_info=True)
    finally:
        console.print("程序已停止", style="bold blue")